package com.shujia.flink.core;

import lombok.*;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.SlidingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

public class Demo6Apply {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStream<String> linesDS = env.socketTextStream("master", 8888);

        DataStream<String> wordsDS = linesDS
                .flatMap((line, out) -> {
                    for (String word : line.split(",")) {
                        out.collect(word);
                    }
                }, Types.STRING);

        //分组划分窗口
        WindowedStream<String, String, TimeWindow> windowDS = wordsDS
                //将相同的key分到一个组内进行计算
                .keyBy(word -> word)
                //将同一个时间段的数据分到一个组内进行计算
                .window(SlidingProcessingTimeWindows.of(Time.seconds(15), Time.seconds(5)));

        //apply: 用于处理窗口之后窗口内数据的方法
        DataStream<ApplyResult> resultDS = windowDS
                .apply(new WindowFunction<String, ApplyResult, String, TimeWindow>() {
                    @Override
                    public void apply(String key,//分组的key
                                      TimeWindow window,//窗口对象，可以获取到窗口的开始和结束时间
                                      Iterable<String> input,//一个窗口内一个key所有的数据
                                      Collector<ApplyResult> out//用于将结果数据发送到下游
                    ) throws Exception {
                        //计算单词的数量
                        long count = 0L;
                        for (String word : input) {
                            count += 1;
                        }
                        //获取窗口的开始和结束时间
                        long start = window.getStart();
                        long end = window.getEnd();
                        //将数据发送到下游
                        out.collect(new ApplyResult(key, count, start, end));
                    }
                });

        resultDS.print();

        env.execute();
    }
}

//lombok插件，在编译时自动生成代码
//@ToString
//@Getter
//@NoArgsConstructor
@Data
@AllArgsConstructor
class ApplyResult {
    String key;
    long count;
    long start;
    long end;
}